package com.iceland.common.rpc;

import com.iceland.common.message.CmdMessage;
import com.iceland.common.message.MessageType;
import com.iceland.common.util.AttributeKeyUtils;
import com.iceland.common.util.ChannelContext;
import io.netty.channel.Channel;
import lombok.extern.slf4j.Slf4j;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * rpc通道以及消息管理
 */
@Slf4j
public class RpcMessageManager {

    //用于rpc通信的通道
    private Channel channel;
    private volatile Map<String,RpcMessage> rpcMessages=new ConcurrentHashMap<>();

    private static final long timeout=1000*10;

    private RpcServiceInterfaceFactory rpcServiceInterfaceFactory=new RpcServiceInterfaceFactory(this);
    private RpcServiceClassFactory rpcServiceClassFactory=new RpcServiceClassFactory();

    public RpcMessageManager() {
    }

    public RpcMessageManager(Channel channel) {
        this.channel=channel;
    }

    public Channel getChannel() {
        return channel;
    }

//    /**
//     * 相互绑定
//     * @param channel
//     */
//    public synchronized void bindChannelToEachOther(Channel channel) {
//
//        if(this.channel!=null && this.channel.isActive()){
//            log.warn("该通道如果不需要使用了请关闭");
//            //this.channel.close();
//        }
//        this.channel = channel;
//        ChannelContext channelContext=AttributeKeyUtils.getChannelContext(channel);
//        channelContext.setRpcMessageManager(this);
//    }


    public void setChannel(Channel channel) {
        this.channel = channel;
    }

    /**
     * 获取通信通道上下文
     * @return
     */
    public  ChannelContext getChannelContext(){
        if(channel==null){
            return null;
        }
        return AttributeKeyUtils.getChannelContext(channel);
    }

    /**
     * 请求调用，同步方式调用
     * @param request
     * @return
     */
    public RpcResponse request(RpcRequest request ){
        RpcMessage rpcMessage=new RpcMessage();
        rpcMessage.setRpcRequest(request);

        rpcMessages.put(request.getRequestId(),rpcMessage);
        sendMessage(request);
        try {
            log.debug("发送请求，阻塞等待响应,id:{}",request.getRequestId());
            long startTime=System.currentTimeMillis();
            synchronized (rpcMessage){
                rpcMessage.wait(timeout);
            }
            log.debug("请求响应时长{},id:{}",System.currentTimeMillis()-startTime,request.getRequestId());
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        RpcResponse response = rpcMessage.getResponse();
        if(response==null){
            response=new RpcResponse();
            response.setRequestId(request.getRequestId());
            response.setRpcStatus(RpcStatusEnum.ERROR);
            response.setException("响应超时,id:"+request.getRequestId());
        }
        //移除
        rpcMessages.remove(request.getRequestId());
        return response;
    }

    /**
     * 添加响应，唤醒被请求阻塞的线程
     * @param response
     */
    public void response(RpcResponse response ){
        RpcMessage rpcMessage = rpcMessages.get(response.getRequestId());
        rpcMessage.setResponse(response);
        //唤醒阻塞的请求
        synchronized (rpcMessage){
            rpcMessage.notifyAll();

        }
    }

    private void sendMessage(RpcRequest request ){
        CmdMessage message=new CmdMessage(MessageType.METHOD_INVOKE.getType(),null,request);
        channel.writeAndFlush(message);
    }

    public RpcServiceInterfaceFactory getRpcServiceInterfaceFactory() {
        return rpcServiceInterfaceFactory;
    }

    public void setRpcServiceInterfaceFactory(RpcServiceInterfaceFactory rpcServiceInterfaceFactory) {
        this.rpcServiceInterfaceFactory = rpcServiceInterfaceFactory;
    }

    public RpcServiceClassFactory getRpcServiceClassFactory() {
        return rpcServiceClassFactory;
    }

    public void setRpcServiceClassFactory(RpcServiceClassFactory rpcServiceClassFactory) {
        this.rpcServiceClassFactory = rpcServiceClassFactory;
    }
}
